package com.banggood.pss.mqtools.bean;

import com.banggood.pss.mqtools.util.ChannelCreator;
import com.rabbitmq.client.*;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;

import java.io.IOException;
import java.util.Map;


/**
 * 用于mq的消息接收
 * @author Mr.chen
 * @date 2020/8/14 14:40
 */
public class JsonMessageReceiver extends MqBaseBean implements InitializingBean, DisposableBean {

    int qos = 10;

    public JsonMessageReceiver(String queueName) {
        super(queueName);
    }

    public JsonMessageReceiver(String queueName, int qos) {
        super(queueName);
        this.qos = qos;
    }

    public JsonMessageReceiver(String queueName,String exchangeName, int qos) {
        super(queueName, exchangeName);
        this.qos = qos;
    }

    public JsonMessageReceiver(String queueName,String exchangeName) {
        super(queueName, exchangeName);
    }

    public JsonMessageReceiver(String queueName,String exchangeName,String routingKey, int qos) {
        super(queueName, exchangeName, routingKey);
        this.qos = qos;
    }

    public JsonMessageReceiver(String queueName,String exchangeName,String routingKey) {
        super(queueName, exchangeName, routingKey);
    }

    public JsonMessageReceiver(String queueName, String exchangeName,String routingKey, BuiltinExchangeType builtinExchangeType) {
        super(queueName, exchangeName, routingKey,builtinExchangeType);
    }

    /**
     * mq消息消费 这里必须同步处理任务，自动ack
     * @param json
     */
    public void syncHandler(String json) {

    }

    /**
     * 本方法是必须要自己手动ack
     * @param consumerTag
     * @param envelope
     * @param properties
     * @param json
     */
    public void handler(String consumerTag, Envelope envelope,
                            AMQP.BasicProperties properties,String json) throws IOException {
        try {
            syncHandler(json);
            getChannel().basicAck(envelope.getDeliveryTag(), false);
        } catch (Exception e) {
            logger.info(e.getMessage(),e);
            //当失败的时候手动Nack，将信息放进死信队列里
            //getChannel().basicNack(envelope.getDeliveryTag(), false, false);
            //直接当做完成的状态，避免消息因为真实的错误无法消费。所以只做错误记录
            getChannel().basicAck(envelope.getDeliveryTag(), false);
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        createQueue();
        start();
    }

    public void start() {
        Consumer consumer = new DefaultConsumer(getChannel()) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body) throws IOException {
                handler(consumerTag, envelope, properties, new String(body,"UTF-8"));
            }
        };
        try {
            getChannel().basicQos(qos);
            //手动确认
            getChannel().basicConsume(queueName, false, consumer);
        } catch (IOException e) {
            logger.info(e.getMessage(),e);
        }
    }

    @Override
    public void destroy() throws Exception {
        channelCreator.closeConnect();
        ChannelCreator.closeChannel(getChannel());
    }
}
